package link.luyu.plugin.brochain.web3;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Scheduler;
import io.reactivex.schedulers.Schedulers;
import java.math.BigInteger;
import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import link.luyu.plugin.brochain.web3.methods.response.Block;
import link.luyu.plugin.brochain.web3.methods.response.Page;
import link.luyu.plugin.brochain.web3.observers.BlockNumberObserver;
import org.web3j.protocol.core.DefaultBlockParameter;

public class Web3Rx {

    private final Web3 web3;
    private final ScheduledExecutorService scheduledExecutorService;
    private final Scheduler scheduler;

    public Web3Rx(Web3 web3, ScheduledExecutorService scheduledExecutorService) {
        this.web3 = web3;
        this.scheduledExecutorService = scheduledExecutorService;
        this.scheduler = Schedulers.from(scheduledExecutorService);
    }

    public Flowable<BigInteger> blockNumberFlowable(long pollingInterval) {
        return Flowable.create(
                subscriber -> {
                    BlockNumberObserver blockNumberObserver =
                            new BlockNumberObserver(web3, subscriber::onNext);
                    blockNumberObserver.run(scheduledExecutorService, pollingInterval);
                    subscriber.setCancellable(blockNumberObserver::cancel);
                },
                BackpressureStrategy.BUFFER);
    }

    public Flowable<Block> blockFlowable(long pollingInterval) {
        return blockNumberFlowable(pollingInterval)
                .flatMap(
                        blockNumber ->
                                web3.brochainGetBlock(DefaultBlockParameter.valueOf(blockNumber))
                                        .flowable())
                .map(
                        block -> {
                            if (block.getBlock().isPresent()) {
                                return block.getBlock().get();
                            } else {
                                throw new Exception("Error fetching block");
                            }
                        });
    }

    public Flowable<Optional<Page>> pageFlowable(String regionId, long pollingInterval) {
        return blockFlowable(pollingInterval)
                .map(
                        block -> {
                            Optional<String> pageHash = Optional.empty();
                            for (Block.RegionPage regionPage : block.getRegionPages()) {
                                if (regionPage.getRegionId().equals(regionId)) {
                                    pageHash = Optional.of(regionPage.getPageHash());
                                    break;
                                }
                            }
                            return pageHash;
                        })
                .flatMap(
                        pageHash -> {
                            Optional<Page> page = Optional.empty();
                            if (pageHash.isPresent()) {
                                return web3.brochainGetPage(regionId, pageHash.get(), true)
                                        .flowable()
                                        .map(
                                                brochainPage -> {
                                                    if (brochainPage.getPage().isPresent()) {
                                                        return brochainPage.getPage();
                                                    } else {
                                                        throw new Exception("Error fetching page");
                                                    }
                                                });
                            } else {
                                return Flowable.just(page);
                            }
                        });
    }
}
